/**
 * 
 */
package com.shelpe.services.sinbad.service.mq.impl;

import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import com.shelpe.services.sinbad.service.base.BaseService;
import com.shelpe.services.sinbad.service.mq.ShelpeMQProducerService;


/**
 * kafka 发消息的实现类
 * @author caidao 
 *
 */
@Service
public class ShelpeMQProducerServiceImpl extends BaseService implements ShelpeMQProducerService {

	/**
	 * 自动注入的kafkaTemplate
	 */
	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;
	
	
	@Value("${kafka.sinbad.homesynctopic}")
	private String homesynctopic;
	
	@Value("${kafka.sinbad.logservertopic}")
	private String logservertopic;
	
	
	
	/**
	 * 发送消息给mq
	 * @param topic topic名称
	 * @param content 内容，我们约定用json格式
	 */
	private void send(String topic, String content) {
//	    logger.info("sending payload='{}' to topic='{}'", content, topic);
	    kafkaTemplate.send(topic, content);
	}


	/**
	 * 当用户登录的时候发送通知，用来同步首页实时数据
	 * @param content
	 */
	@Override
	public void sendMsgForLoginSync(String content) {
		if ( StringUtils.isNotBlank(content) ) {
			this.send(homesynctopic, content);
		}
		else {
			logger.error("====== sendMsgForLoginSync mq content is null");
		}		
	}

	/**
	 * 当优化时候发送通知，用来同步首页实时数据
	 * @param content
	 */
	@Override
	public void sendMsgForOptimize(String content) {
		if ( StringUtils.isNotBlank(content) ) {
			this.send(homesynctopic, content);
		}
		else {
			logger.error("====== sendMsgForOptimize mq content is null");
		}
	}


//	@Override
//	public void sendKeywordsLog(String content) {
//		if ( StringUtils.isNotBlank(content) ) {
//			this.send(logservertopic, content);
//		}
//		else {
//			logger.error("====== sendKeywordsLog mq content is null");
//		}		
//	}	
}
